package com.sample.springbatch.job.sample.mybatiscursor;

import com.sample.springbatch.config.BaseConfig;
import com.sample.springbatch.mybatis.entity.RepayOrderInfoEntity;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.batch.MyBatisCursorItemReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.SynchronizedItemStreamReader;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author zwg
 * @date 2025/2/11 17:30
 */
@Configuration
public class MybatisCursorJobConfig extends BaseConfig {

    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    Logger logger = LoggerFactory.getLogger(MybatisCursorJobConfig.class);

    public static ConcurrentHashMap<Long,Integer> DATA_MAP = new ConcurrentHashMap<>();

    // 注解@JobScope必须添加，才能在reader 中获取到参数
    @Bean(name = "mybatisCursorMultiThreadJob")
    public Job mybatisCursorMultiThreadJob() {
        return jobBuilderFactory.get("mybatisCursorMultiThreadJob")
                .incrementer(new RunIdIncrementer())
                .listener(myJobListener) //增加对job的监听
                .start(cleanDataMapStep())
                .next(mybatisCursorMultiThreadJobStep())
                .next(checkDataMapStep())
                .build();
    }

    @Bean(name="mybatisCursorMultiThreadJobStep")
    public Step mybatisCursorMultiThreadJobStep() {
        return stepBuilderFactory.get("mybatisCursorMultiThreadJobStep")
                //.listener()  //添加对step的监听，可以实现多step的所有监听
                .<RepayOrderInfoEntity, RepayOrderInfoEntity> chunk(500)   //这些配置参数需要单独按job配置到一个公共的数据类中，支持通过application.yml 调整修改
                //跳过机制是针对每个执行线程设置，最大跳过数量为：skipLimit * throttleLimit
                .reader(getMyBatisCursorItemReader())
                .processor(getMyBatisCursorItemProcessor())
                .writer(new ItemWriter() {
                    @Override
                    public void write(List items) throws Exception {

                    }
                })
                .taskExecutor(new SimpleAsyncTaskExecutor())
                .throttleLimit(2) ////这些配置参数需要单独按job配置到一个公共的数据类中，支持通过application.yml 调整修改
                .build();
    }


    @Bean
    public SynchronizedItemStreamReader getMyBatisCursorItemReader() {
        MyBatisCursorItemReader reader = new MyBatisCursorItemReader();
        reader.setQueryId("com.sample.springbatch.mybatis.mapper.RepayOrderInfoMapper.getRepayOrderInfos");
        reader.setSqlSessionFactory(sqlSessionFactory);
        //使用多线程 必须是同步读取
        SynchronizedItemStreamReader synchronizedItemStreamReader = new SynchronizedItemStreamReader();
        synchronizedItemStreamReader.setDelegate(reader);
        return synchronizedItemStreamReader;
    }

    @Bean
    public ItemProcessor getMyBatisCursorItemProcessor(){
        return new ItemProcessor<RepayOrderInfoEntity, RepayOrderInfoEntity>() {
            @Override
            public RepayOrderInfoEntity process(RepayOrderInfoEntity item) throws Exception {
                //logger.info("id:{}",item.getId());
                if(MybatisCursorJobConfig.DATA_MAP.containsKey(item.getId())){
                    MybatisCursorJobConfig.DATA_MAP.put(item.getId(), MybatisCursorJobConfig.DATA_MAP.get(item.getId())+1);
                }else{
                    MybatisCursorJobConfig.DATA_MAP.put(item.getId(),1);
                }
                return item;
            }
        };
    }

    @Bean
    public Step cleanDataMapStep() {
        return stepBuilderFactory.get("cleanDataMapStep")
                .tasklet(new Tasklet(){
                    @Override
                    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                        MybatisCursorJobConfig.DATA_MAP.clear();
                        logger.info("cleanDataMapStep clear ok");
                        return RepeatStatus.FINISHED;
                    }
                })
                .build();
    }

    @Bean
    public Step checkDataMapStep() {
        return stepBuilderFactory.get("checkDataMapStep")
                .tasklet(new Tasklet(){
                    @Override
                    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                        boolean checkResult = true;
                        for(Long id : MybatisCursorJobConfig.DATA_MAP.keySet()){
                            if(MybatisCursorJobConfig.DATA_MAP.get(id)>1){
                                checkResult = false;
                                logger.info("id {} times :{}",id,MybatisCursorJobConfig.DATA_MAP.get(id));
                            }
                        }
                        logger.info("checkDataMapStep check {} ok",MybatisCursorJobConfig.DATA_MAP.size());
                        if(!checkResult){
                            throw new RuntimeException("checkDataMapStep fail");
                        }
                        return RepeatStatus.FINISHED;
                    }
                })
                .build();
    }
}
